package producer.broker;

import api.Message;
import api.MessageType;
import api.exception.MessageRunTimeException;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import common.convert.GenericMessageConvert;
import common.convert.RabbitMessageConvert;
import common.serializer.Serializer;
import common.serializer.SerializerFactory;
import common.serializer.impl.JacksonSerializer;
import common.serializer.impl.JacksonSerializerFactory;
import org.checkerframework.checker.units.qual.A;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.function.Predicate;

/**
 * @Desc: "RabbitTemplate 做池化封装"
 * 好处：
 *      每一个topic 对应一个RabbitTemplate
 *      1. 提高发送效率
 *      2. 可以根据不同的需求指定化不同的RabbitTemplate，，比如每一个topic 都有自己的routingkey规则
 * @Author: caixiang
 * @DATE: 2020/7/8 16:39
 */

@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback {

    static Logger logger = LoggerFactory.getLogger(RabbitTemplateContainer.class);

    private Splitter splitter = Splitter.on("#");
    private Map<String /* topic */, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();

    @Autowired
    private ConnectionFactory connectionFactory;

    private SerializerFactory serializerFactory = JacksonSerializerFactory.INSTANCE;

    //RabbitTemplate 是发送消息的模板
    public RabbitTemplate getTemplate(Message message)throws MessageRunTimeException {
        Preconditions.checkNotNull(message);
        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);
        if(rabbitTemplate != null){
            return rabbitTemplate;
        }
        logger.info("#RabbitTemplateContainer.getTemplate# topic:{} is not exits,create one",topic);
        RabbitTemplate newRabbitTemplate = new RabbitTemplate(connectionFactory);
        newRabbitTemplate.setExchange(topic);
        newRabbitTemplate.setRetryTemplate(new RetryTemplate());

        //指定Message 为转换类型
        Serializer serializer = serializerFactory.create(message.getClass());
        //api.message 和 amqp.message 之间的一个转换器
        GenericMessageConvert genericMessageConvert = new GenericMessageConvert(serializer);
        //api.message 装饰类
        RabbitMessageConvert rabbitMessageConvert = new RabbitMessageConvert(genericMessageConvert);

        //对于message 序列化方式
        newRabbitTemplate.setMessageConverter(rabbitMessageConvert);


        String messageType = message.getMessageType();

        //判断type 的目的是可以  依据不同的 message类型 自定义的发送一些自定义数据
        if(!MessageType.RAPID.equals(messageType)){
            //因为this 本身就是ConfirmCallback ，所以当你发的是 MessageType.CONFIRM 消息，会自动 附载 上confirm() 的自动回调方法
            newRabbitTemplate.setConfirmCallback(this);
            //newRabbitTemplate.setChannelTransacted(true);  支持事务
        }
        //如果不存在的 放入
        rabbitMap.putIfAbsent(topic,newRabbitTemplate);

        rabbitMap.put(topic,newRabbitTemplate);
        return rabbitMap.get(topic);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //具体的消息应答，，只要不是迅速消息 任何消息 在对方收到的时候都要做一次callback（表明我收到了）

        List<String> strings = splitter.splitToList(correlationData.getId());
        String messageId = strings.get(0);
        Long sendTime = Long.parseLong(strings.get(1));
        //如果是ack = true 了那么说明 broker已经落盘成功
        if(ack){
            //1.为true 就说明，，投到MQ 成功， 就依据messageId 更新数据库里面的状态 为 SEND_OK就行了

            //这边后续写 相关的业务 代码，，一般是落库到数据库
            logger.info("send message is ok,confirm messageId:{},send time :{}",messageId,sendTime);
        }else {
            //如果失败的时候 ，，先不更新，，直到重试最大次数，，在置为失败
            logger.info("send message is fail,confirm messageId:{},send time :{}",messageId,sendTime);
        }
    }
}
